package prictice.transaction1;

import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.transactional.ITransactionalSpout;
import org.apache.storm.tuple.Fields;

import java.util.HashMap;
import java.util.Map;
import java.util.Random;

/**
 * @Package prictice.transaction1
 * @Description:
 * @Author elwyn
 * @Date 2017/9/16 19:45
 * @Email elonyong@163.com
 */
public class MyTxSpout implements ITransactionalSpout<MyMate> {

    private static final long serialVersionUID = -7720388478246437004L;
    Map<Long, String> dbMap = null;

    public MyTxSpout() {
        dbMap = new HashMap<>();
        Random random = new Random();
        String[] host = {"www.taobao.com"};
        String[] sessionId = {"HJKHKJHUIFDISH21312321", "JKHIHNIDFSFDS34535", "FDSFDSFDSFDSMKMK908", "DFSFDSFDSAAAA1123245", "GFGDGFDGHKINNIU778"};
        String[] time = {"2014-01-07 08:40:50", "2014-01-07 08:40:50",
                "2014-02-07 06:40:50", "2014-01-27 06:40:50",
                "2014-03-07 09:40:50", "2014-01-17 11:40:50",
                "2014-04-07 08:40:50", "2014-11-08 12:40:50",
                "2014-05-07 08:40:50"};
        for (long i = 0; i < 50; i++) {
            dbMap.put(i, host[0] + "\t" + sessionId[random.nextInt(5)] + "\t" + time[random.nextInt(9)]);

        }
    }


    @Override
    public Coordinator<MyMate> getCoordinator(Map map, TopologyContext topologyContext) {
        return new MyCoordinator();
    }

    @Override
    public Emitter<MyMate> getEmitter(Map map, TopologyContext topologyContext) {
        return new MyEmitter(dbMap);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("tx", "log"));
    }

    @Override
    public Map<String, Object> getComponentConfiguration() {
        return null;
    }
}
